package com.navinfo.data.esper.cep.linkid;


import com.espertech.esper.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class LinkidMainPatternRecogThread {
    private static final Logger log = LoggerFactory.getLogger(LinkidMainPatternRecogThread.class);

    public static class MyRunnable implements Runnable {
        private final CountDownLatch latch;
        private final Map<Integer, List<Integer>> map;
        private final EPRuntime runtime;

        public MyRunnable(CountDownLatch latch, Map<Integer, List<Integer>> map, EPRuntime runtime) {
            this.latch = latch;
            this.map = map;
            this.runtime = runtime;
        }

        public void run() {
            log.info("map.size:"+map.size());
            for (int i = 701100; i < 901108; i++) {
                for (int j = 0; j < map.get(i).size(); j++) {
//                    log.info("map.get(i).size:"+map.get(i).size());
                    TerminalLink tl = new TerminalLink();
                    tl.setTid(i);
                    tl.setLinkid(map.get(i).get(j));
                    runtime.sendEvent(tl);
                }
            }
            latch.countDown();
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        // Configure engine with event names to make the statements more readable.
        // This could also be done in a configuration file.
        Configuration configuration = new Configuration();
        configuration.getEngineDefaults().getThreading()
                .setListenerDispatchPreserveOrder(false);
        configuration.getEngineDefaults().getThreading()
                .setInternalTimerEnabled(false);   // remove thread that handles time advancing
        configuration.addEventType("TerminalLink", TerminalLink.class.getName());
        // Get engine instance
        EPServiceProvider epService = EPServiceProviderManager.getProvider("linkid", configuration);

        epService.getEPAdministrator().getConfiguration().setMatchRecognizeMaxStates(100L);

        log.info("making test data begin......");
        //创建测试数据集
        Map<Integer, List<Integer>> map = new HashMap<Integer, List<Integer>>();
        for (int i = 701100; i < 901108; i++) {
            List<Integer> li = new ArrayList<Integer>();
            for (int j = 1; j < 240; j++) {
                li.add(j+901099-i);
            }
            map.put(i,li);
        }

        List<Integer> roadLinkids=new ArrayList<Integer>();
        for (int j = 1; j < 4; j++) {
            roadLinkids.add(200000-1+j);
        }

        log.info("making test data end......");

        List<Integer> tidList=new ArrayList<Integer>();

//        String pattern = "select * from TerminalLink "
//                + "match_recognize ("
//                + "measures A.tid as tid "
//                + "pattern (A B* C D* E) "
//                + "define "
//                + "A as A.linkid=1,"
//                + "B as B.tid=A.tid,"
//                + "C as C.tid=A.tid and C.linkid=2,"
//                + "D as D.tid=A.tid,"
//                + "E as E.tid=A.tid and E.linkid=3)";

        StringBuffer pattern = new StringBuffer().append("select * from TerminalLink match_recognize ( partition by tid measures A.tid as tid after match skip to next row ");
        StringBuffer _sub_pattern = new StringBuffer().append(" pattern ( ");
        StringBuffer _sub_define = new StringBuffer().append(" define ");
        int roadlinkid = 0;
        for(int z=0;z<roadLinkids.size();z++){
            roadlinkid = roadLinkids.get(z);
            if (z==0) {
                _sub_pattern.append(" A ").append(" B0* ");
                _sub_define.append(" A as A.linkid=").append(roadlinkid).append(", B0 as B0.tid=A.tid, ");
            } else if (z==(roadLinkids.size()-1)) {
                _sub_pattern.append(" A").append(z).append(") ");
                _sub_define.append(" A").append(z).append(" as A").append(z).append(".linkid=").append(roadlinkid)
                        .append(" and A").append(z).append(".tid=A.tid)");
            } else {
                _sub_pattern.append(" A").append(z).append(" B").append(z).append("* ");
                _sub_define.append(" A").append(z).append(" as A").append(z).append(".linkid=").append(roadlinkid)
                        .append(" and A").append(z).append(".tid=A.tid, ")
                        .append(" B").append(z).append(" as B").append(z).append(".tid=A.tid, ");
            }
        }

        String epl = pattern.append(_sub_pattern).append(_sub_define).toString();
        log.info(epl);
        EPStatement statement = epService.getEPAdministrator().createEPL(epl);
        LinkidPatternListenerThread linkidPatternListenerThread = new LinkidPatternListenerThread();
        statement.addListener(linkidPatternListenerThread);

        int numThreads = 1;
        Thread[] threads = new Thread[numThreads];
        CountDownLatch latch = new CountDownLatch(numThreads);


        for (int i = 0; i < numThreads; i++) {
            threads[i] = new Thread(
                    new MyRunnable(latch, map, epService.getEPRuntime()));
        }
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < numThreads; i++) {
            threads[i].start();
        }

        latch.await(10, TimeUnit.MINUTES);
        if (latch.getCount() > 0) {
            throw new RuntimeException("Failed to complete in 10 minute");
        }
        long delta = System.currentTimeMillis() - startTime;
        System.out.println("Took " + delta + " millis");

        if ((linkidPatternListenerThread.getEventlist()!=null)&&(linkidPatternListenerThread.getEventlist().size()>0)) {
            Iterator<EventBean> it2 = linkidPatternListenerThread.getEventlist().iterator();
            while (it2.hasNext()) {
//                log.info(it2.next() + " ");
                tidList.add((int)it2.next().get("tid"));
            }
        }

        List<Integer> unique = tidList.stream().distinct().collect(Collectors.toList());
        Iterator<Integer> it2 = unique.iterator();
        while (it2.hasNext()) {
            log.info(it2.next() + " ");
        }

        epService.destroy();
    }
}
